package com.flx.dis.mq.rocket.service.producer;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author: Fenglixiong
 * @Date: 2021/6/29 18:48
 * @Description:
 *
 * 每次推送消息会执行executeLocalTransaction方法，首先会发送半消息，到这里的时候是执行具体本地业务，
 * 执行成功后手动返回RocketMQLocalTransactionState.COMMIT状态，这里是保证本地事务执行成功，
 * 如果本地事务执行失败则可以返回ROLLBACK进行消息回滚。此时消息只是被保存到broker，并没有发送到topic中，
 * broker会根据本地返回的状态来决定消息的处理方式。
 *
 * 监听事务消息
 *
 * 1.发送方向 MQ 服务端发送消息。
 * 2.MQ Server 将消息持久化成功之后，向发送方 ACK 确认消息已经发送成功，此时消息为半消息。
 * 3.发送方开始执行本地事务逻辑。
 * 4.发送方根据本地事务执行结果向 MQ Server 提交二次确认（Commit 或是 Rollback），MQ Server 收到 Commit 状态则将半消息标记为可投递，订阅方最终将收到该消息；MQ Server 收到 Rollback 状态则删除半 消息，订阅方将不会接受该消息。
 * 5.在断网或者是应用重启的特殊情况下，上述步骤4提交的二次确认最终未到达 MQ Server，经过固定时间后 MQ Server 将对该消息发起消息回查。
 * 6.发送方收到消息回查后，需要检查对应消息的本地事务执行的最终结果。
 * 7.发送方根据检查得到的本地事务的最终状态再次提交二次确认，MQ Server 仍按照步骤4对半消息进行操作。
 *
 */
@Component
@RocketMQTransactionListener
public class TransactionListener implements RocketMQLocalTransactionListener {

    private static Map<String, RocketMQLocalTransactionState> STATE_MAP = new HashMap<>();

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        String transId = (String)message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        try{
            System.out.println("执行本地事务");
            Thread.sleep(500);
            // 设置事务状态
            STATE_MAP.put(transId, RocketMQLocalTransactionState.COMMIT);
            // 返回事务状态给生产者
            return RocketMQLocalTransactionState.COMMIT;
        }catch (Exception e){
            e.printStackTrace();
        }
        STATE_MAP.put(transId, RocketMQLocalTransactionState.ROLLBACK);
        return RocketMQLocalTransactionState.ROLLBACK;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        try {
            String transId = (String)message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
            System.out.println("回查消息 -> transId = " + transId + ", state = " + STATE_MAP.get(transId));
            return STATE_MAP.get(transId);
        }catch (Exception e){
            e.printStackTrace();
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }

}
